package cn.tecnova.utils

import java.util.concurrent.Future

import org.apache.kafka.clients.producer.{KafkaProducer, ProducerRecord, RecordMetadata}

/**
  * description:自定义kafka生产者避免来一条消息new一次
  **/
class BroadcastKafkaProducer[K,V](createproducer:() => KafkaProducer[K,V]) extends Serializable{

  lazy val producer = createproducer()
  def send(topic:String,key:K,value:V):Future[RecordMetadata] = producer.send(new ProducerRecord[K,V](topic,key,value))

}

object BroadcastKafkaProducer  {

  import scala.collection.JavaConversions._
  def apply[K,V](config:Map[String,Object]):BroadcastKafkaProducer [K,V] = {
    val createProducerFunc  = () => {
      val producer = new KafkaProducer[K,V](config)
      sys.addShutdownHook({
        producer.close()
      })
      producer
    }
    new BroadcastKafkaProducer (createProducerFunc)
  }
  def apply[K, V](config: java.util.Properties): BroadcastKafkaProducer[K, V] = apply(config.toMap)

}